/*
 * FaceCat图形通讯框架
 * 著作权编号:2015SR229355+2020SR0266727
 * 上海卷卷猫信息技术有限公司
 */

#include "..\\..\\stdafx.h"
#include "..\\..\\include\\service\\FCClientService.h"
#include "..\\..\\include\\core\\FCBinary.h"
#include "..\\..\\include\\sock\\FCClientSockets.h"
#include "..\\..\\..\\zlib\\zlib.h"

namespace FaceCat{
    long long m_downFlowClient = 0;

    long long m_upFlowClient = 0;

    FCLock m_clientDataCallBacksLock;

    int gZCompress(Byte *data, uLong ndata, Byte *zdata, uLong *nzdata){
        z_stream c_stream;
        int err = 0;

        if(data && ndata > 0) {
            c_stream.zalloc = Z_NULL;
            c_stream.zfree = Z_NULL;
            c_stream.opaque = Z_NULL;
            if(deflateInit2(&c_stream, Z_DEFAULT_COMPRESSION, Z_DEFLATED,
                MAX_WBITS + 16, 6, Z_DEFAULT_STRATEGY) != Z_OK) {
                return -1;
            }

            c_stream.next_in  = data;
            c_stream.avail_in  = ndata;
            c_stream.next_out = zdata;
            c_stream.avail_out  = *nzdata;
            while(c_stream.avail_in != 0 && c_stream.total_out < *nzdata) {
                if(deflate(&c_stream, Z_NO_FLUSH) != Z_OK) {
                    return -1;
                }
            }

            if(c_stream.avail_in != 0) {
                return c_stream.avail_in;
            }

            for(;;) {
                if((err = deflate(&c_stream, Z_FINISH)) == Z_STREAM_END){
                    break;
                }
                if(err != Z_OK){
                    return -1;
                }
            }

            if(deflateEnd(&c_stream) != Z_OK){
                return -1;
            }
            *nzdata = c_stream.total_out;
            return 0;
        }
        return -1;
    }

    int gZDeCompress(Byte *zdata, uLong nzdata, Byte *data, uLong *ndata){
        int err = 0;
        z_stream d_stream = {0}; /* decompression stream */
        static char dummy_head[2] = {
            0x8 + 0x7 * 0x10,
            (((0x8 + 0x7 * 0x10) * 0x100 + 30) / 31 * 31) & 0xFF,
        };

        d_stream.zalloc = Z_NULL;
        d_stream.zfree = Z_NULL;
        d_stream.opaque = Z_NULL;
        d_stream.next_in  = zdata;
        d_stream.avail_in = 0;
        d_stream.next_out = data;
        if(inflateInit2(&d_stream, MAX_WBITS + 16) != Z_OK) {
            return -1;
        }

        while(d_stream.total_out < *ndata && d_stream.total_in < nzdata) {
            d_stream.avail_in = d_stream.avail_out = 1; /* force small buffers */
            if((err = inflate(&d_stream, Z_NO_FLUSH)) == Z_STREAM_END){
                break;
            }
            if(err != Z_OK) {
                if(err == Z_DATA_ERROR){
                    d_stream.next_in = (Bytef*) dummy_head;
                    d_stream.avail_in = sizeof(dummy_head);

                    if((err = inflate(&d_stream, Z_NO_FLUSH)) != Z_OK){
                        return -1;
                    }
                } else
                {
                    return -1;
                }
            }
        }
        if(inflateEnd(&d_stream) != Z_OK){
            return -1;
        }
        *ndata = d_stream.total_out;
        return 0;
    }

    ArrayList<FCDataEventCallBack*> m_clientDataCallBacks;

    FCMessage::FCMessage(){
            m_serviceID = 0;
            m_functionID = 0;
            m_requestID = 0;
            m_socketID = 0;
            m_state = 0;
            m_compressType = 0;
            m_bodyLength = 0;
            m_body = 0;
            m_signal = 0;
        }

        FCMessage::FCMessage(int serviceID, int functionID, int requestID, int socketID, int state, int compressType, int bodyLength, char *body){
            m_serviceID = serviceID;
            m_functionID = functionID;
            m_requestID = requestID;
            m_socketID = socketID;
            m_state = state;
            m_compressType = compressType;
            m_bodyLength = bodyLength;
            m_body = body;
            m_signal = 0;
        }

        ///////////////////////////////////////////////////////////////////////////////////////////////////////

        FCMessageListener::FCMessageListener(){
        }

        FCMessageListener::~FCMessageListener(){
            clear();
        }

        void FCMessageListener::add(FCListenerMessageCallBack *callBack, Object pInvoke){
            m_callBacks.add(callBack);
            m_callBackInvokes.add(pInvoke);
        }

        void FCMessageListener::callBack(FCMessage *message){
            int callBacksSize = (int)m_callBacks.size();
            for(int i = 0; i < callBacksSize; i++)
            {
                m_callBacks.get(i)->callListenerMessageEvent((Object)m_callBacks.get(i), message, m_callBackInvokes.get(i));
            }
        }

        void FCMessageListener::clear(){
            m_callBacks.clear();
            m_callBackInvokes.clear();
        }

        void FCMessageListener::remove(FCListenerMessageCallBack *callBack){
            for(int i = 0; i < m_callBacks.size(); i++){
                if(callBack == m_callBacks.get(i)){
                    m_callBacks.removeAt(i);
                    m_callBackInvokes.removeAt(i);
                    break;;
                }
            }
        }


    MessageCallBack m_messageCallBackClient;

    ArrayList<FCClientService*> m_servicesClient;

    WriteLogCallBack m_writeLogCallBackClient;

    int FCClientService::closeClient(int socketID){
        return FCClientSockets::close(socketID);
    }

    int FCClientService::connectToServer(int proxyType, const char *ip, int port, const char *proxyIp, int proxyPort, const char *proxyUserName, const char *proxyUserPwd, const char *proxyDomain, int timeout, char *token){
        return FCClientSockets::connect(proxyType, ip, port, proxyIp, proxyPort, proxyUserName, proxyUserPwd, proxyDomain, timeout, token);
    }

    void FCClientService::registerCallBack(MessageCallBack callBack){
        return FCClientSockets::registerRecv(callBack);
    }

    void FCClientService::registerLog(WriteLogCallBack callBack){
        return FCClientSockets::registerLog(callBack);
    }

    int FCClientService::sendByClient(int socketID, const char *str, int len){
        return FCClientSockets::send(socketID, str, len);
    }

    FCClientService::FCClientService(){
        m_compressType = COMPRESSTYPE_NONE;
        m_serviceID = 0;
    }

    FCClientService::~FCClientService(){
        m_lock.lock();
        m_listeners.clear();
        m_waitMessages.clear();
        m_lock.unLock();
    }

    int FCClientService::getCompressType(){
        return m_compressType;
    }

    void FCClientService::setCompressType(int value){
        m_compressType = value;
    }

    long long FCClientService::getDownFlow(){
        return m_downFlowClient;
    }

    void FCClientService::setDownFlow(long long value){
        m_downFlowClient = value;
    }

    int FCClientService::getServiceID(){
        return m_serviceID;
    }

    void FCClientService::setServiceID(int value){
        m_serviceID = value;
    }

    long long FCClientService::getUpFlow(){
        return m_upFlowClient;
    }

    void FCClientService::setUpFlow(long long value){
        m_upFlowClient = value;
    }

    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

    void FCClientService::addDataCallBack(FCDataEventCallBack *dataCallBack){
        m_clientDataCallBacksLock.lock();
        m_clientDataCallBacks.add(dataCallBack);
        m_clientDataCallBacksLock.unLock();
    }

    void FCClientService::addService(FCClientService *service){
        if (!m_messageCallBackClient){
            MessageCallBack messageCallBack = &callBack;
            m_messageCallBackClient = messageCallBack;
            registerCallBack(messageCallBack);
            WriteLogCallBack writeLogCallBack = &writeServerLog;
            m_writeLogCallBackClient = writeLogCallBack;
            registerLog(m_writeLogCallBackClient);
        }
        m_servicesClient.add(service);
    }

    void FCClientService::callBack(int socketID, int localSID, const char *str, int len){
        m_downFlowClient += len;
        if(m_clientDataCallBacks.size() > 0){
            m_clientDataCallBacksLock.lock();
            for(int i = 0; i < m_clientDataCallBacks.size(); i++){
                if(m_clientDataCallBacks.get(i)->callDataEvent(socketID, localSID, str, len)){
                    m_clientDataCallBacksLock.unLock();
                    return;
                }
            }
            m_clientDataCallBacksLock.unLock();
        }
        if (len > 4){
            FCBinary binary;
            binary.write(str, len);
            int head = binary.readInt();
            int serviceID = (int)binary.readShort();
            int macthCount = 0;
            for(int i = 0; i < m_servicesClient.size(); i++){
                FCClientService *service = m_servicesClient.get(i);
                if(service->getServiceID() == serviceID){
                    if(macthCount > 0){
                        FCBinary brNew;
                        brNew.write(str, len);
                        brNew.readInt();
                        brNew.readShort();
                        service->onCallBack(&brNew, socketID, localSID, len);
                    }else{
                        service->onCallBack(&binary, socketID, localSID, len);
                    }
                    macthCount++;
                }
            }
        }
    }

    int FCClientService::getRequestID(){
        return m_requestID ++;
    }

    void FCClientService::getServices(ArrayList<FCClientService*> *services){
        for(int i = 0; i < m_servicesClient.size(); i++){
            services->add(m_servicesClient.get(i));
        }
    }

    int FCClientService::keepAlive(int socketID){
        FCBinary binary;
        binary.initialize(4, false);
        binary.writeInt(4);
        char *str = (char*)binary.getBytes();
        return sendByClient(socketID, str, 4);
    }

    void FCClientService::onCallBack(FCBinary *binary, int socketID, int localSID, int len){
        int headSize = sizeof(int) * 3 + sizeof(short) * 2 + sizeof(char) * 2;
        int functionID = (int)binary->readShort();
        int requestID = binary->readInt();
        int state = (int)binary->readChar();
        int compressType = (int)binary->readChar();
        int uncBodyLength = binary->readInt();
        int bodyLength = len - headSize;
        char *body = new char[bodyLength];
        memset(body, '\0', bodyLength);
        binary->read(body, bodyLength);
        if (compressType == COMPRESSTYPE_GZIP){
            uLong deLen = uncBodyLength;
            char *buffer = new char[deLen];
            memset(buffer, '\0', deLen);
            gZDeCompress((Byte*)body, bodyLength, (Byte*)buffer, &deLen);
            delete[] body;
            body = buffer;
        }
        FCMessage message(getServiceID(), functionID, requestID, socketID, state, compressType, uncBodyLength, body);
        onReceive(&message);
        onWaitMessageHandle(&message);
        if(message.m_body){
            delete[] body;
        }
        body = 0;
    }

    void FCClientService::onClientClose(int socketID, int localSID){
        m_lock.lock();
        for(int i = 0; i < m_waitMessages.size(); i++){
            FCMessage *message = m_waitMessages.getValue(i);
            if(message->m_socketID == socketID){
                message->m_signal = 1;
            }
        }
        m_lock.unLock();
    }

    void FCClientService::onClientConnect(int socketID, int localSID){
    }

    void FCClientService::onReceive(FCMessage *message){
    }

    void FCClientService::onWaitMessageHandle(FCMessage *message){
        if((int)m_waitMessages.size() > 0){
            m_lock.lock();
            if(m_waitMessages.containsKey(message->m_requestID))
            {
                FCMessage *waitMessage = m_waitMessages.get(message->m_requestID);
                waitMessage->copy(message);
                char *newBody = new char[message->m_bodyLength];
                for (int i = 0; i < message->m_bodyLength; i++){
                    newBody[i] = message->m_body[i];
                }
                waitMessage->m_body = newBody;
            }
            m_lock.unLock();
        }
    }

    void FCClientService::addListener(int requestID, FCListenerMessageCallBack *callBack, Object pInvoke){
        m_lock.lock();
        FCMessageListener *listener = 0;
        if(m_listeners.containsKey(requestID))
        {
            listener = m_listeners.get(requestID);
        }
        else
        {
            listener = new FCMessageListener;
            m_listeners.put(requestID, listener);
        }
        listener->add(callBack, pInvoke);
        m_lock.unLock();
    }

    void FCClientService::addWait(int requestID, FCMessage *message){
        m_lock.lock();
        m_waitMessages.put(requestID, message);
        m_lock.unLock();
    }

    int FCClientService::send(FCMessage *message){
        FCBinary binary;
        char *buffer = 0;
        const char *body = message->m_body;
        int bodyLength = message->m_bodyLength;
        int uncBodyLength = bodyLength;
        if (message->m_compressType == COMPRESSTYPE_GZIP){
            uLong cLen = 128;
            if(uncBodyLength > cLen)
            {
                cLen = uncBodyLength;
            }
            buffer = new char[cLen];
            memset(buffer, '\0', cLen);
            gZCompress((Byte*)body, bodyLength, (Byte*)buffer, &cLen);
            body = buffer;
            bodyLength = (int)cLen;
        }
        int len = sizeof(int) * 3 + bodyLength + sizeof(short) * 2 + sizeof(char) * 2;
        binary.initialize(len, false);
        binary.writeInt(len);
        binary.writeShort((short)message->m_serviceID);
        binary.writeShort((short)message->m_functionID);
        binary.writeInt(message->m_requestID);
        binary.writeChar((char)message->m_state);
        binary.writeChar((char)message->m_compressType);
        binary.writeInt(uncBodyLength);
        binary.write(body, bodyLength);
        char *str = (char*)binary.getBytes();
        if(buffer){
            delete[] buffer;
            buffer = 0;
        }
        int ret = sendByClient(message->m_socketID, str, len);
        m_upFlowClient += ret;
        return ret;
    }

    void FCClientService::sendToListener(FCMessage *message){
        FCMessageListener *listener = 0;
        m_lock.lock();
        if(m_listeners.containsKey(message->m_requestID))
        {
            listener = m_listeners.get(message->m_requestID);
        }
        m_lock.unLock();
        if(listener){
            listener->callBack(message);
        }
    }

    void FCClientService::removeDataCallBack(FCDataEventCallBack *dataCallBack){
        m_clientDataCallBacksLock.lock();
        int size = m_clientDataCallBacks.size();
        for(int i = 0; i < size; i++){
            if(m_clientDataCallBacks.get(i) == dataCallBack){
                m_clientDataCallBacks.removeAt(i);
                i--;
                size--;
            }
        }
        m_clientDataCallBacksLock.unLock();
    }

    void FCClientService::removeListener(int requestID){
        m_lock.lock();
        if(m_listeners.containsKey(requestID))
        {
            m_listeners.remove(requestID);
        }
        m_lock.unLock();
    }

    void FCClientService::removeListener(int requestID, FCListenerMessageCallBack *callBack){
        m_lock.lock();
        if(m_listeners.containsKey(requestID))
        {
            m_listeners.get(requestID)->remove(callBack);
        }
        m_lock.unLock();
    }

    void FCClientService::removeWait(int requestID){
        m_lock.lock();
        if(m_waitMessages.containsKey(requestID))
        {
            m_waitMessages.remove(requestID);
        }
        m_lock.unLock();
    }

    int FCClientService::waitMessage(int requestID, int timeout){
        int state = 0;
        while(timeout > 0){
            m_lock.lock();
            if(m_waitMessages.containsKey(requestID))
            {
                FCMessage *message = m_waitMessages.get(requestID);
                if(message->m_bodyLength > 0){
                    state = 1;
                    m_lock.unLock();
                    break;
                }else if(message->m_signal == 1){
                    break;
                }
            }
            else{
                m_lock.unLock();
                break;
            }
            m_lock.unLock();
            timeout -= 10;
            Sleep(10);
        }
        removeWait(requestID);
        return state;
    }

    void FCClientService::writeServerLog(int socketID, int localSID, int state, const char *log){
        if (state == 2 || state == 3) {
            for(int i = 0; i < m_servicesClient.size(); i++){
                FCClientService *service = m_servicesClient.get(i);
                service->onClientClose(socketID, localSID);
            }
        } else if (state == 1) {
            for(int i = 0; i < m_servicesClient.size(); i++){
                FCClientService *service = m_servicesClient.get(i);
                service->onClientConnect(socketID, localSID);
            }
        }
    }
}
